package org.snail.proxy.server.netty;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.channel.*;
import io.netty.channel.epoll.EpollEventLoopGroup;
import io.netty.channel.epoll.EpollServerSocketChannel;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.util.concurrent.DefaultThreadFactory;
import io.netty.util.concurrent.Future;
import io.netty.util.concurrent.FutureListener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.snail.proxy.server.DSystem;
import org.snail.proxy.server.config.NettyConfig;

import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;

public abstract class NettyServer {
    private static final Logger log = LoggerFactory.getLogger(NettyServer.class);

    private final ServerBootstrap bootstrap = new ServerBootstrap();
    private final NettyConfig nettyConfig;

    private EventLoopGroup bossGroup;
    private EventLoopGroup workerGroup;

    public NettyServer(NettyConfig nettyConfig) {
        this.nettyConfig = nettyConfig;
    }

    public void start(
            Consumer<Future<Void>> bindCallBack,
            final Consumer<Future<Void>> activeCallBack,
            Runnable runnable
    ) {
        startAsync(bindCallBack, runnable).syncUninterruptibly().addListener((FutureListener<Void>) future -> {
            if (activeCallBack != null) {
                activeCallBack.accept(future);
            }
        });
    }

    public Future<Void> startAsync(final Consumer<Future<Void>> callBack, Runnable runnable) {
        initGroups(runnable);

        Class<? extends ServerChannel> channelClass = NioServerSocketChannel.class;
        if (!DSystem.isWindows()) {
            channelClass = EpollServerSocketChannel.class;
        }

        bootstrap.group(bossGroup, workerGroup)
                .channel(channelClass)
                .childHandler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel sh) {
                        NettyServer.this.initChannel(sh);
                    }
                });
        applyConnectionOptions(bootstrap);
        return bootstrap.bind(nettyConfig.getIp(), nettyConfig.getPort()).addListener((FutureListener<Void>) future -> {
            log.info("bind {}:{}", nettyConfig.getIp(), nettyConfig.getPort());
            if (callBack != null) {
                callBack.accept(future);
            }
        });
    }

    protected abstract void initChannel(SocketChannel sh);

    private void initGroups(Runnable runnable) {
        if (!DSystem.isWindows()) {
            bossGroup = new EpollEventLoopGroup(getBossThreadNum(), new DefaultThreadFactory("BOSS_LOOP"));
            workerGroup = new EpollEventLoopGroup(getWorkerThreadNum(), new DefaultThreadFactory("WORKER_LOOP"));
        } else {
            bossGroup = new NioEventLoopGroup(getBossThreadNum(), new DefaultThreadFactory("BOSS_LOOP"));
            workerGroup = new NioEventLoopGroup(getWorkerThreadNum(), new DefaultThreadFactory("WORKER_LOOP"));
        }
        for (int i = 0; i < getWorkerThreadNum(); i++) {
            EventLoop next = workerGroup.next();
            next.scheduleAtFixedRate(runnable, 0, getFlushInterval(), TimeUnit.MILLISECONDS);
        }
    }

    private void applyConnectionOptions(ServerBootstrap bootstrap) {
        bootstrap.option(ChannelOption.SO_REUSEADDR, true);
        bootstrap.option(ChannelOption.SO_BACKLOG, getTcpSoBackLog());
        bootstrap.option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);

        bootstrap.childOption(ChannelOption.SO_REUSEADDR, true);
        bootstrap.childOption(ChannelOption.SO_KEEPALIVE, true);
        bootstrap.childOption(ChannelOption.TCP_NODELAY, true);
        bootstrap.childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);
        bootstrap.childOption(ChannelOption.RCVBUF_ALLOCATOR, new AdaptiveRecvByteBufAllocator());
    }

    private int getTcpSoBackLog() {
        return nettyConfig.getTcplog();
    }

    private int getBossThreadNum() {
        return nettyConfig.getBossNum();
    }

    private int getWorkerThreadNum() {
        return nettyConfig.getWorkerNum();
    }

    private int getFlushInterval() {
        return nettyConfig.getFlushInterval();
    }

    public void close() {
        bossGroup.shutdownGracefully();
        workerGroup.shutdownGracefully();
        log.info("Close proxy {}.", nettyConfig);
    }

    public abstract void onSendMessage(String id);
}
